package cc.jinglupeng.kom.controller;

import cc.jinglupeng.kom.bean.Result;
import cc.jinglupeng.kom.bean.ZkBroker;
import cc.jinglupeng.kom.bean.ConsumerTopicPartitionState;
import cc.jinglupeng.kom.data.KafkaData;
import cc.jinglupeng.kom.data.ZkData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;

/**
 * Created by max357 on 2016/4/11.
 */
@RestController
@RequestMapping("/data")
public class DataController {

    private static final Logger logger = LoggerFactory.getLogger(DataController.class);

    @RequestMapping("/brokers.json")
    public String getAllBrokers() {
        try {
            ZkBroker[] brokers = ZkData.getAllBrokers();
            return Result.newSucceededInstance(brokers).toString();
        } catch (Exception e) {
            logger.error("get all brokers failed.", e);
            return Result.newFailedInstance(-1, e.getMessage()).toString();
        }
    }

    @ResponseBody
    @RequestMapping("/consumers.json")
    public String getAllConsumers() {
        try {
            List<String> consumers = ZkData.getAllConsumer();
            return Result.newSucceededInstance(consumers).toString();
        } catch (Exception e) {
            logger.error("get all consumers failed.", e);
            return Result.newFailedInstance(-1, e.getMessage()).toString();
        }
    }

    @ResponseBody
    @RequestMapping("/topics.json")
    public String getAllTopics() {
        try {
            List<String> topics = ZkData.getAllTopic();
            return Result.newSucceededInstance(topics).toString();
        } catch (Exception e) {
            logger.error("get all topics failed.", e);
            return Result.newFailedInstance(-1, e.getMessage()).toString();
        }
    }

    @ResponseBody
    @RequestMapping("/{consumer}/topics.json")
    public String getConsumerTopics(@PathVariable("consumer") String consumer) {
        try {
            List<String> topics = ZkData.getConsumerTopics(consumer);
            return Result.newSucceededInstance(topics).toString();
        } catch (Exception e) {
            logger.error("get all topics failed.", e);
            return Result.newFailedInstance(-1, e.getMessage()).toString();
        }
    }

    @ResponseBody
    @RequestMapping("/{consumer}/{topic}/partitions.json")
    public String getConsumerTopicPartitionsStates(@PathVariable("consumer") String consumesr, @PathVariable("topic") String topic) {
        try {
            ConsumerTopicPartitionState[] partitionStates = KafkaData.getConsumerTopicPartitionStates(consumesr, topic);
            return Result.newSucceededInstance(partitionStates).toString();
        } catch (Exception e) {
            logger.error("get all topics failed.", e);
            return Result.newFailedInstance(-1, e.getMessage()).toString();
        }
    }
}
